#include "lyj_kv_store.h"
#include "lyj_log.h"
#include "lyj_filenode.h"
#include "rocksdb/c.h"
#include <io.h>
#include <fcntl.h>


struct lyj_kv_store_s
{
	char file_path[1024];
	rocksdb_t* db;
	rocksdb_options_t* db_options;
	HANDLE mutex;
#if !LYJ_HEADER_BUFFER_KV
	HANDLE head_handle;
	int head_fd;
#endif
};

lyj_inline static void lyj_check_and_mkdir(const char* path)
{
	DWORD attrib = GetFileAttributesA(path);
	if (attrib == INVALID_FILE_ATTRIBUTES || !(attrib & FILE_ATTRIBUTE_DIRECTORY))
	{
		if (!CreateDirectoryA(path, NULL))
		{
			lyj_real_log("CreateDirectory %s error %x", path, GetLastError());
			exit(4);
		}
	}
}
lyj_kv_store_t* lyj_kv_store_init(LPCWSTR store_path)
{
	char db_path[1024];
	lyj_memzero(db_path, sizeof(char) * 1024);

	lyj_kv_store_t* kv = lyj_type_malloc(lyj_kv_store_t);
	lyj_type_memzero(kv);

	wcstombs(db_path, store_path, 1024);
	_snprintf(kv->file_path, 1024, "%s\\file", db_path);
	lyj_memcpy(db_path + strlen(db_path), "\\meta", 5);
	lyj_check_and_mkdir(kv->file_path);
	lyj_check_and_mkdir(db_path);
#if !LYJ_HEADER_BUFFER_KV
	char head_file_path[1024];

	_snprintf(head_file_path, 1024, "%s\\head_file", kv->file_path);
	kv->head_handle = CreateFileA(head_file_path,
		FILE_GENERIC_READ | FILE_GENERIC_WRITE,
		FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
		NULL,
		OPEN_ALWAYS,
		0,
		NULL);
	if (kv->head_handle == INVALID_HANDLE_VALUE)
	{
		lyj_real_log("open %s error %d", head_file_path, GetLastError());
		exit(-1);
		return NULL;
	}
	kv->head_fd = _open_osfhandle((intptr_t)kv->head_handle, O_BINARY | O_CREAT | O_RDWR);
#endif
	rocksdb_options_t *options = rocksdb_options_create();
	rocksdb_block_based_table_options_t *block_based_table_options = rocksdb_block_based_options_create();

	
	//rocksdb_options_set_db_write_buffer_size(options, 2 << 20);
	//rocksdb_options_set_target_file_size_base(options, 2 * 1048576);
	//rocksdb_options_set_max_bytes_for_level_base(options, 10 * 1048576);
	//rocksdb_options_set_soft_pending_compaction_bytes_limit(options, 256 * 1048576);
	//rocksdb_options_set_hard_pending_compaction_bytes_limit(options, 1073741824ul);
	//rocksdb_block_based_options_set_cache_index_and_filter_blocks(block_based_table_options, 1);
	//rocksdb_block_based_options_set_index_type(block_based_table_options, 2);

	//rocksdb_filterpolicy_t *filterpolicy = rocksdb_filterpolicy_create_bloom(10);
	rocksdb_cache_t* block_cache = rocksdb_cache_create_lru(1024 * 1024 * 1024);

	rocksdb_block_based_options_set_block_size(block_based_table_options, 16 * 1024);

	rocksdb_block_based_options_set_data_block_index_type(block_based_table_options, 1);
	rocksdb_block_based_options_set_data_block_hash_ratio(block_based_table_options, 0.75);

	//rocksdb_block_based_options_set_filter_policy(block_based_table_options, filterpolicy);
	rocksdb_block_based_options_set_block_cache(block_based_table_options, block_cache);

	//rocksdb_options_set_memtable_prefix_bloom_size_ratio(options, 0.02);
	//rocksdb_options_set_memtable_whole_key_filtering(options, 1);
	rocksdb_options_set_block_based_table_factory(options, block_based_table_options);

	rocksdb_options_optimize_level_style_compaction(options, 512 * 1024 * 1024);
	rocksdb_options_increase_parallelism(options, 4);
	// create the DB if it's not already present
	rocksdb_options_set_create_if_missing(options, 1);

	char* err = NULL;
	kv->mutex = CreateMutex(NULL, FALSE, NULL);
	kv->db = rocksdb_open(options, db_path, &err);
	rocksdb_cache_destroy(block_cache);
	//rocksdb_filterpolicy_destroy(filterpolicy);
	rocksdb_block_based_options_destroy(block_based_table_options);
	rocksdb_options_destroy(options);
	if (err)
	{
		lyj_real_log("rocksdb_open error %s", err);
		free(err);
		exit(2);
	}
	return kv;
}

const char* lyj_kv_store_get_file_path(lyj_kv_store_t* kv)
{
	return kv->file_path;
}
const int lyj_kv_store_get_head_file_fd(lyj_kv_store_t* kv)
{
#if !LYJ_HEADER_BUFFER_KV
	return kv->head_fd;
#else
	return -1;
#endif
}
const HANDLE lyj_kv_store_get_head_file_handle(lyj_kv_store_t* kv)
{
#if !LYJ_HEADER_BUFFER_KV
	return kv->head_handle;
#else
	return NULL;
#endif
}
lyj_ret_t lyj_kv_store_put(lyj_kv_store_t* kv, const lyj_kv_buf_t* key, const lyj_kv_buf_t* value)
{
	rocksdb_writeoptions_t* wo = rocksdb_writeoptions_create();
	char* err = NULL;
	rocksdb_put(kv->db, wo, key->buf, key->len, value->buf, value->len, &err);
	rocksdb_writeoptions_destroy(wo);
	if (err != NULL)
	{
		lyj_real_log("rocksdb_put error %s", err);
		free(err);
		return LYJ_RET_ERROR;
	}
	return LYJ_RET_OK;
}

void lyj_kv_store_range_prefix(lyj_kv_store_t* kv, const lyj_kv_buf_t* prefix_key, void* arg, lyj_kv_range_callback_f callback)
{
	if (!callback) return;
	rocksdb_readoptions_t* ro = rocksdb_readoptions_create();
	rocksdb_iterator_t* iter = rocksdb_create_iterator(kv->db, ro);
	lyj_kv_buf_t key = lyj_kv_buf_null;
	lyj_kv_buf_t value = lyj_kv_buf_null;
	rocksdb_iter_seek(iter, prefix_key->buf, prefix_key->len);
	while (rocksdb_iter_valid(iter))
	{
		key.buf = (void *)rocksdb_iter_key(iter, &key.len);
		if (key.len < prefix_key->len || 0 != memcmp(key.buf, prefix_key->buf, prefix_key->len)) break;
		value.buf = (void *)rocksdb_iter_value(iter, &value.len);
		callback(arg, &key, &value);
		rocksdb_iter_next(iter);
		
	}
	rocksdb_iter_destroy(iter);
	rocksdb_readoptions_destroy(ro);
}

lyj_bool_t lyj_kv_store_range_prefix_exist(lyj_kv_store_t* kv, const lyj_kv_buf_t* prefix_key)
{
	rocksdb_readoptions_t* ro = rocksdb_readoptions_create();
	rocksdb_iterator_t* iter = rocksdb_create_iterator(kv->db, ro);
	lyj_kv_buf_t key = lyj_kv_buf_null;
	lyj_bool_t ret = LYJ_FALSE;
	rocksdb_iter_seek(iter, prefix_key->buf, prefix_key->len);
	if (rocksdb_iter_valid(iter))
	{
		key.buf = (void*)rocksdb_iter_key(iter, &key.len);
		if (key.len >= prefix_key->len && 0 == memcmp(key.buf, prefix_key->buf, prefix_key->len))
		{
			ret = LYJ_TRUE;
		}
	}
	rocksdb_iter_destroy(iter);
	rocksdb_readoptions_destroy(ro);
	return ret;
}

lyj_ret_t lyj_kv_store_get(lyj_kv_store_t* kv, const lyj_kv_buf_t* key, lyj_kv_buf_t* value)
{
	rocksdb_readoptions_t* ro = rocksdb_readoptions_create();
	char* err = NULL;
	value->buf = rocksdb_get(kv->db, ro, key->buf, key->len, &value->len, &err);
	rocksdb_readoptions_destroy(ro);
	if (err != NULL)
	{
		lyj_real_log("rocksdb_get error %s", err);
		free(err);
		return LYJ_RET_ERROR;
	}
	return value->buf ? LYJ_RET_OK : LYJ_RET_ERROR;
}

lyj_ret_t lyj_kv_store_delete(lyj_kv_store_t* kv, const lyj_kv_buf_t* key)
{

	rocksdb_writeoptions_t* wo = rocksdb_writeoptions_create();
	char* err = NULL;
	rocksdb_delete(kv->db, wo, key->buf, key->len, &err);
	rocksdb_writeoptions_destroy(wo);
	if (err != NULL)
	{
		lyj_real_log("rocksdb_delete error %s", err);
		free(err);
		return LYJ_RET_ERROR;
	}
	return LYJ_RET_OK;
}
void lyj_kv_store_destroy(lyj_kv_store_t* kv)
{
	if (!kv) return;
	if (kv->db) rocksdb_close(kv->db);
	if (kv->mutex) CloseHandle(kv->mutex);
	lyj_free(kv);
}

void lyj_kv_store_lock(lyj_kv_store_t* kv)
{
	WaitForSingleObject(kv->mutex, INFINITE);
}

void lyj_kv_store_unlock(lyj_kv_store_t* kv)
{
	ReleaseMutex(kv->mutex);
}
